package xyz.syyrjx.consumer.client;

import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.util.concurrent.DefaultPromise;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import xyz.syyrjx.consumer.conf.SyyrjxRpcConsumerConfig;
import xyz.syyrjx.request.SyyrjxRpcCallRequest;
import xyz.syyrjx.request.SyyrjxRpcGetAddressRequest;
import xyz.syyrjx.client.SyyrjxRpcClient;
import xyz.syyrjx.consumer.client.exception.SyyrjxRpcConsumerException;
import xyz.syyrjx.consumer.entity.SyyrjxRpcFuture;
import xyz.syyrjx.consumer.entity.SyyrjxRpcFutureHolder;
import xyz.syyrjx.consumer.handler.SyyrjxRpcConsumerClientHandler;
import xyz.syyrjx.entity.SyyrjxRpcRemoteAddress;
import xyz.syyrjx.initializer.SyyrjxRpcChannelInitializer;
import xyz.syyrjx.util.SyyrjxRpcRandomUtil;

import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

/**
 * @author 大地崩坏苍蝇兽
 * @date 2023/1/8 - 19:48
 */
@NoArgsConstructor
@AllArgsConstructor
@Data
public class SyyrjxRpcConsumerClient {

    private final static Logger LOGGER = LoggerFactory.getLogger(SyyrjxRpcConsumerClient.class);

    private SyyrjxRpcClient client = new SyyrjxRpcClient();

    private Map<SyyrjxRpcRemoteAddress, SyyrjxRpcClient> clientMap = new ConcurrentHashMap<>();

    @Autowired
    SyyrjxRpcConsumerConfig config;

    /**
     * 客户端远程调用提供者
     *
     * @param provider    提供者唯一标识
     * @param serviceName 服务唯一标识
     * @param args        请求参数
     * @return
     * @throws ExecutionException
     * @throws InterruptedException
     */
    public <T> T remoteCall(String provider, String serviceName, long timeout, Object... args) throws ExecutionException, InterruptedException, TimeoutException {
        while (client.getChannel() == null || !client.getChannel().isActive()) {
            this.start();
        }
        SyyrjxRpcClient clientGetResult = null;
        //从注册中心获取提供者路径
        if (config.getRegistryAddress().length > 0) {
            Set<SyyrjxRpcRemoteAddress> addresses = getProviderAddress(provider);
            //提供者全部宕机
            if (addresses.size() == 0) {
                throw new SyyrjxRpcConsumerException(SyyrjxRpcConsumerException.NO_PROVIDER);
            }
            LOGGER.debug(addresses.toString());

            int randomIndex = SyyrjxRpcRandomUtil.randomInt(addresses.size());
            SyyrjxRpcRemoteAddress address = (SyyrjxRpcRemoteAddress) addresses.toArray()[randomIndex];

            LOGGER.debug("需要远程调用的地址为" + address);

            clientGetResult = clientMap.get(address);
            LOGGER.debug("从远程访问客户端表中取出的客户端：" + clientGetResult);
            if (clientGetResult == null) {
                clientGetResult = clientMap.putIfAbsent(address,
                        getClientTOGetResult(address.getHost(), address.getPort()));
                if (clientGetResult == null) {
                    clientGetResult = clientMap.get(address);
                }
                LOGGER.debug("新建的远程访问客户端：" + clientGetResult);
                SyyrjxRpcConsumerClientHandler handler = new SyyrjxRpcConsumerClientHandler();
                SyyrjxRpcChannelInitializer channelInitializer = new SyyrjxRpcChannelInitializer();
                channelInitializer.setHandler(handler);
                channelInitializer.setMaxReadSize(config.getMaxReadSize());
                clientGetResult.run(channelInitializer);
            }
        }
        //从配置文件获取提供者路径
        if (clientGetResult == null) {
            clientGetResult = client;
        }
        LOGGER.debug("远程调用的客户端为" + clientGetResult);
        return getRemoteResult(clientGetResult, provider, serviceName, timeout, args);
    }

    public SyyrjxRpcClient getClientTOGetResult(String remoteHost, int remotePort) throws InterruptedException {
        SyyrjxRpcClient client = new SyyrjxRpcClient();
        client.setRemoteHost(remoteHost);
        client.setRemotePort(remotePort);
        return client;
    }

    /**
     * 从注册中心获取地址
     *
     * @param provider 提供者唯一标识
     * @param <T>      SyyrjxRpcRemoteAddress类型，远程路径
     * @return SyyrjxRpcRemoteAddress类型，远程路径
     * @throws ExecutionException   throws
     * @throws InterruptedException throws
     */
    public <T> T getProviderAddress(String provider) throws ExecutionException, InterruptedException {
        SyyrjxRpcGetAddressRequest request = new SyyrjxRpcGetAddressRequest();
        request.setProviderName(provider);
        request.setFutureId(SyyrjxRpcFutureHolder.FUTURE_ID.getAndIncrement());
        LOGGER.debug("futureId为：" + request.getFutureId());

        SyyrjxRpcFuture<T> future = new SyyrjxRpcFuture<>(new DefaultPromise(new NioEventLoopGroup().next()));
        SyyrjxRpcFutureHolder.FUTURE_MAP.put(request.getFutureId(), future);
        client.send(request);
        LOGGER.debug("等待返回注册中心消息");
        return future.getPromise().get();
    }

    /**
     * 远程调用提供者返回结果
     *
     * @param provider    提供者唯一标识
     * @param serviceName 服务接口唯一标识
     * @param args        参数
     * @param <T>         结果类型
     * @return 调用结果
     * @throws ExecutionException   throws
     * @throws InterruptedException throws
     */
    public <T> T getRemoteResult(SyyrjxRpcClient client, String provider, String serviceName, long timeout,Object... args) throws ExecutionException, InterruptedException, TimeoutException {
        SyyrjxRpcCallRequest request = new SyyrjxRpcCallRequest();
        request.setMagic(0xabcd);
        request.setProviderName(provider);
        request.setFutureId(SyyrjxRpcFutureHolder.FUTURE_ID.getAndIncrement());
        LOGGER.debug("futureId为：" + request.getFutureId());

        SyyrjxRpcFuture<T> future = new SyyrjxRpcFuture<>(new DefaultPromise(new NioEventLoopGroup().next()));
        SyyrjxRpcFutureHolder.FUTURE_MAP.put(request.getFutureId(), future);

        request.setMethodName(serviceName);

        request.setParam(args);

        client.send(request);
        if (timeout == 0) {
            return future.getPromise().get();
        }
        return future.getPromise().get(timeout, TimeUnit.MILLISECONDS);
    }

    /**
     * 启动客户端连接服务端
     *
     * @throws InterruptedException throw
     */
    public void start() throws InterruptedException {
        LOGGER.debug("配置类为" + config);
        String[] registryAddressArray = config.getRegistryAddress();
        List<String> registryAddressList = new ArrayList<>(Arrays.asList(registryAddressArray));
        int currentRegistryIndex = -1;
        boolean connectSuccess = false;
        while (!connectSuccess) {
            try {
                if (registryAddressArray.length == 0) {
                    LOGGER.debug("连接提供者");
                    client.setRemoteHost(config.getRemoteHost());
                    client.setRemotePort(config.getRemotePort());
                } else {
                    if (registryAddressList.size() == 0) {
                        throw new SyyrjxRpcConsumerException("连接注册中心失败");
                    }
                    LOGGER.debug("连接注册中心");
                    currentRegistryIndex = SyyrjxRpcRandomUtil.randomInt(registryAddressList.size());
                    String[] registryAddress =
                            registryAddressList.get(currentRegistryIndex)
                                    .split(":");
                    String registryHost = registryAddress[0];
                    int registryPort = Integer.parseInt(registryAddress[1]);
                    client.setRemoteHost(registryHost);
                    client.setRemotePort(registryPort);
                }
                LOGGER.debug("准备连接" + client);
                SyyrjxRpcConsumerClientHandler handler = new SyyrjxRpcConsumerClientHandler();
                SyyrjxRpcChannelInitializer channelInitializer = new SyyrjxRpcChannelInitializer();
                channelInitializer.setHandler(handler);
                channelInitializer.setMaxReadSize(config.getMaxReadSize());
                client.run(channelInitializer);
                connectSuccess = true;
            } catch (Exception e) {
                registryAddressList.remove(currentRegistryIndex);
                LOGGER.error("连接失败,尝试重连", e);
                if (e instanceof SyyrjxRpcConsumerException) {
                    throw e;
                }
            }
        }
    }
}
